package cn._51doit.flink.day07;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SideOutputDemo1 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        //1
        //a
        //2
        //奇数的标签
        OutputTag<Integer> oddTag = new OutputTag<Integer>("oddTag"){};
        //偶数的标签
        OutputTag<Integer> evenTag = new OutputTag<Integer>("evenTag"){};
        //字符串标签
        OutputTag<String> strTag = new OutputTag<String>("strTag"){};

        //经过侧流输出，主流中的数据按照是否打标签有两种类型的数据（一种打标签，一种未打标签）
        SingleOutputStreamOperator<String> mainStream = lines.process(new ProcessFunction<String, String>() {

            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                try {
                    int num = Integer.parseInt(value);
                    if (num % 2 != 0) {
                        //奇数
                        ctx.output(oddTag, num); //输出奇数并关联对应的标签
                    } else {
                        //偶数
                        ctx.output(evenTag, num); //输出偶数并关联对应的标签
                    }
                } catch (NumberFormatException e) {
                    //e.printStackTrace();
                    //字符串
                    ctx.output(strTag, value);
                }
                //将全部的数据都输出（未打标签的）
                out.collect(value);
            }
        });

        //打印主流中的数据
        mainStream.print("main "); //打印未打标签的

        //从主流中获取指定标签的侧流
        DataStream<Integer> oddStream = mainStream.getSideOutput(oddTag);

        DataStream<String> strStream = mainStream.getSideOutput(strTag);

        oddStream.print("odd ");
        strStream.print("str ");

        env.execute();

    }

}
